#include "Upload.pb.h"
#include "UploadService.h"
#include "CryptoUtil.h"
#include "MessageQueue.h"

#include <signal.h>
#include <nlohmann/json.hpp>
#include <workflow/WFFacilities.h>
#include <workflow/MySQLResult.h>
#include <workflow/WFTaskFactory.h>
#include <ppconsul/agent.h>
#include <string>
#include <vector>
#include <fstream>

using namespace srpc;
using namespace protocol;
using namespace ppconsul::agent;
using std::string;
using std::vector;
using std::ofstream;
using ppconsul::Consul;

namespace
{

	using json = nlohmann::json;

	WFFacilities::WaitGroup wait_group(1);


	//保存前端上传文件的路径
	constexpr const char* DEFAULT_FILE_SAVE_PATH = "./resources/";
	//oss对象的基本路径名
	constexpr const char* DEFAULT_FILE_OSS_BACKUP_PATH = "resources/";
	//默认的注册服务器的端口
	constexpr const unsigned short DEFAULT_UPLOAD_PORT = 1238;
	//默认的注册服务器的ip
	constexpr const char* DEFAULT_UPLOAD_IP = "127.0.0.1";
	//mysql的url
	constexpr const char* MYSQL_URL = "mysql://root:123@127.0.0.1/CloudDisk2";
	//最大重试次数
	constexpr const int RE_TRY_MAX = 3;

	void sig_handler(int signo)
	{
		wait_group.done();
	}


	/* ************************************************************************************ */
	/* ********************定时发送心跳包************************************************* */
	/* ************************************************************************************ */

	void timer_callback(WFTimerTask* task)
	{
		Agent* agent = (Agent*)task->user_data;
		agent->servicePass("Upload1");

		WFTimerTask* next = WFTaskFactory::create_timer_task(7, 0, timer_callback);
		next->user_data = task->user_data;

		series_of(task)->push_back(next);

	}


	/* ************************************************************************************ */
	/* **************************************上传****************************************** */
	/* ************************************************************************************ */

	void handleUpload(::l::UploadResponse* response, string fileContent, string hash_code, WFMySQLTask* task) {

		// 任务失败: 及早失败原则
		if (task->get_state() != WFT_STATE_SUCCESS) {
			response->set_statuscode(500);
#ifdef BUG
			std::cerr << "Error: " << WFGlobal::get_error_string(task->get_state(), task->get_error());
#endif
			return;
		}

		MySQLResponse* resp = task->get_resp();
		MySQLResultCursor cursor{ resp };

		//检查第一个insert哈希值到files的结果集，成功表示文件没有保存，失败表示文件已经保存
		if (cursor.get_cursor_status() == MYSQL_STATUS_OK && cursor.get_affected_rows() > 0) {

#ifdef BUG
			std::cout << "file:" << hash_code << "is not exist need save file" << "\n";
#endif

			string real_path{ DEFAULT_FILE_SAVE_PATH };
			real_path.append(hash_code);
			ofstream ofs{ real_path };

			if (!ofs) {
				response->set_statuscode(500);
#ifdef BUG
				std::cout << real_path << ": failed open" << "\n";
#endif
				return;
			}

			if (!ofs.write(fileContent.c_str(), fileContent.size())) {
				response->set_statuscode(500);
#ifdef BUG
				std::cout << real_path << ": failed write" << "\n";
#endif
				return;
			}


			//oss备份
			std::string objectName{ DEFAULT_FILE_OSS_BACKUP_PATH + hash_code };
			json message{ {"filePath", real_path}, {"objectName", objectName} };
			l::MessageQueue::produce_message(message);
#ifdef BUG
			std::cout << "oss success:" << objectName << "\n";
#endif    

		}
		else {
#ifdef BUG
			std::cout << "file is  exist not save file" << hash_code << "\n";
#endif
		}



		//检查第二个insert的结果集
		cursor.next_result_set();


		if (cursor.get_cursor_status() == MYSQL_STATUS_OK) {

			response->set_statuscode(200);
#ifdef BUG
			std::cout << "mysql insert into users_files success" << "\n";
#endif
		}
		else {

			//上传失败代表后端出错了
			response->set_statuscode(500);
#ifdef BUG
			std::cout << "mysql insert into users_files failed" << "\n";
#endif
		}

	}

}




class UploadServiceImpl : public ::l::Upload::Service
{
public:

	void Upload(::l::UploadRequest* request, ::l::UploadResponse* response, srpc::RPCContext* ctx) override
	{
		string username = request->username();
		string filename = request->filename();
		string fileContent = request->filecontent();

		//计算哈希值，查找并插入file表
		string hash_code{};
		if (!l::CryptoUtil::hashFileContent(hash_code, fileContent)) {
			response->set_statuscode(500);
			return;
		}

#ifdef BUG
		std::cout << "filename: " << filename << "\n";
		std::cout << "fileContent: " << fileContent << "\n";
		std::cout << "hashcaode: " << hash_code << "\n";
#endif

		//在files表中直接插入，如果唯一键冲突，就将size改成size(相当于没改),这是为了让insert操作一定成功
		//因为wfrest中如果某条语句失败了，后面的语句不会继续执行，直接判断第一个结果集，如果更改的行数为0说明已经有这个文件了
		//不需要保存和备份，如果更改行数大于0说明没有保存过这个文件
		//需要插入（并保存文件和备份文件）
		//users_files表中insert记录,在唯一键冲突时，将status重新置为0，并更新last_uptate(由于status可能就是0，
		//所以手动更新last_uptate)

		//拼成一个sql,遍历两个结果集
		string url{ MYSQL_URL };
		//插入files
		string sql{ "INSERT INTO files (filehash, size) VALUES(\"" };
		sql.append(hash_code).append("\",").append(std::to_string(fileContent.size())).append(") ")
			.append("ON DUPLICATE KEY UPDATE size = ").append(std::to_string(fileContent.size())).append(";");

		//插入user_files
		sql.append("INSERT INTO users_files (uid, fid, filename) SELECT users.id,files.id,");
		sql.append("\"").append(filename).append("\"")
			.append(" FROM users CROSS JOIN files WHERE users.username = ")
			.append("\"").append(username).append("\" AND users.tomb = 0 AND files.filehash = \"")
			.append(hash_code).append("\" ON DUPLICATE KEY UPDATE "
				"fid = VALUES(fid),"
				"last_update = NOW(), ")
			.append("status = 0");

#ifdef BUG
		std::cout << "upload sql: " << sql << "\n";
#endif

		WFMySQLTask* task = WFTaskFactory::create_mysql_task(MYSQL_URL, RE_TRY_MAX, std::bind(handleUpload, response,
			std::move(fileContent), std::move(hash_code), std::placeholders::_1));

		task->get_req()->set_query(sql);

		SeriesWork* series = ctx->get_series();
		series->push_back(task);
	}

};

int main()
{

	signal(SIGINT, sig_handler);

	SRPCServer server;

	UploadServiceImpl upload_impl;
	server.add_service(&upload_impl);

	if (server.start(DEFAULT_UPLOAD_PORT) == 0) {

		// 向注册中心注册服务
		Consul consul{ "http://127.0.0.1:8500", ppconsul::kw::dc = "dc1" };
		Agent agent{ consul };

		// 注册服务 
		agent.registerService(
			kw::id = "Upload1",
			kw::name = "Upload",
			kw::address = "127.0.0.1",
			kw::port = DEFAULT_UPLOAD_PORT,
			kw::check = TtlCheck(std::chrono::seconds{ 10 })
		);

		// 定时发送心跳包
		WFTimerTask* timerTask = WFTaskFactory::create_timer_task(7, 0, timer_callback);
		timerTask->user_data = &agent;

		timerTask->start();
		wait_group.wait();
		server.stop();
	}
	else {
#ifdef BUG
		std::cout << "upload server start failed\n";
#endif
	}
	google::protobuf::ShutdownProtobufLibrary();
	return 0;
}
